package com.liunian.heartbeatdetect;

import com.liunian.NettyBootstrapInitializer;
import com.liunian.WrpcBootstrap;
import com.liunian.compress.CompressorFactory;
import com.liunian.discovery.Registry;
import com.liunian.serialize.SerializerFactory;
import com.liunian.transport.message.WrpcRequest;
import com.liunian.wrpcenum.RequestType;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 心跳探测的核心目的是什么？探活，感知那些服务器的连接是正常的，哪些是不正常的。
 */
@Slf4j
public class HeartbeatDetector {

    public static void detectHeartbeat(String serviceName) {
        // 从注册中心拉取服务列表并建立连接
        Registry registry = WrpcBootstrap.getInstance().getConfiguration().getRegistryConfig().getRegistry();
        List<InetSocketAddress> addresses = registry.lookup(serviceName,WrpcBootstrap.getInstance().getConfiguration().getGroup());

        // 将连接进行缓存
        for (InetSocketAddress address : addresses) {
            try {
                if (!WrpcBootstrap.CHANNEL_CACHE.containsKey(address)) {
                    Channel channel = NettyBootstrapInitializer.getBootstrap().connect(address).sync().channel();
                    WrpcBootstrap.CHANNEL_CACHE.put(address, channel);
                }

            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        // 任务，定期发送消息
        Thread thread = new Thread(() -> {
            new Timer().scheduleAtFixedRate(new MyTimerTask(), 0, 2000);
        },"wrpc-HeartbeatDetector-thread");
        thread.setDaemon(true);
        thread.start();

    }

    private static class MyTimerTask extends TimerTask {

        @Override
        public void run() {
            // 将响应时长的map清空
            WrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.clear();
            // 遍历所有的channel
            Map<InetSocketAddress, Channel> cache = WrpcBootstrap.CHANNEL_CACHE;
            for (Map.Entry<InetSocketAddress, Channel> entry : cache.entrySet()) {
                // 定义一个重试的次数
                int tryTimes = 3;
                while (tryTimes > 0) {
                    Channel channel = entry.getValue();
                    long start = System.currentTimeMillis();
                    // 创建一个心跳请求
                    WrpcRequest wrpcRequest = WrpcRequest.builder()
                            .requestId(WrpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId())
                            .compressType(CompressorFactory.getCompressor(WrpcBootstrap.getInstance().getConfiguration().getCompressType()).getCode())
                            .requestType(RequestType.HEART_BEAT.getId())
                            .serializeType(SerializerFactory.getSerializer(WrpcBootstrap.getInstance().getConfiguration().getSerializeType()).getCode())
                            .timeStamp(start)
                            .build();

                    // 四、写出报文
                    CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                    // 将 completableFuture 暴露出去
                    WrpcBootstrap.PENDING_REQUEST.put(wrpcRequest.getRequestId(), completableFuture);
                    channel.writeAndFlush(wrpcRequest).addListener((ChannelFutureListener) promise -> {
                        if (!promise.isSuccess()) {
                            completableFuture.completeExceptionally(promise.cause());
                        }
                    });
                    Long endTime = 0L;
                    try {
                        // 阻塞方法，get()方法如果得不到结果，就会一直阻塞
                        // 我们想不一直阻塞可以参加参数
                        completableFuture.get(1, TimeUnit.SECONDS);
                        endTime = System.currentTimeMillis();
                    } catch (InterruptedException | ExecutionException | TimeoutException e) {
                        // 一旦发生问题，需要优先重试
                        tryTimes--;
                        if (tryTimes == 0) {

                        }
                        log.error("和地址为[{}]的主机连接发生异常，正在进行第[{}]次重试......", channel.remoteAddress(), 3 - tryTimes);
                        // 将重试的机会用尽，将失效的地址移出服务列表。
                        if (tryTimes == 0) {
                            WrpcBootstrap.CHANNEL_CACHE.remove(entry.getKey());
                        }
                        // 尝试等待一段时间后重试
                        try {
                            Thread.sleep(10*(new Random().nextInt(5)));
                        } catch (InterruptedException interruptedException) {
                            throw new RuntimeException(interruptedException);
                        }
                        continue;
                    }
                    Long time = endTime - start;
                    // 使用treeMap进行缓存
                    WrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.put(time, channel);
                    log.debug("和[{}]服务器的响应时间是[{}]。", entry.getKey(), time);
                    break;
                }
            }
            log.info("------------------------------响应时间的treemap-----------------------------");
            for (Map.Entry<Long, Channel> entry : WrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.entrySet()) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}] ---> channelId:[{}]",entry.getKey(), entry.getValue().id());
                }
            }
        }
    }

}
